package rpc

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"rpc/codec"
	"strings"
	"sync"
	"time"
)

// Call 用来保存每一个call请求所需的信息
type Call struct {
	//请求序列号
	Seq uint64
	//请求服务的方法
	ServiceMethod string
	//传递的参数
	Args interface{}
	//应答参数
	Reply interface{}
	//错误信息
	Error error
	//用来控制是不是完成
	Done chan *Call
}

func (call *Call) done()  {
	call.Done <- call
}

// Client 双锁可以提高效率
type Client struct {
	//处理编码以及IO
	cc codec.Codec
	//编码格式
	opt *Option
	//传输数据的io锁
	sending sync.Mutex
	header codec.Header
	//保护共享内存的锁
	mu sync.Mutex
	//当前可用序号
	seq uint64
	//存储没有处理完的请求
	pending map[uint64]*Call
	//用户正常关闭
	closing bool
	//发生错误而关闭
	shutdown bool
}

var ErrShutDown = errors.New("connection is shut down")

func (client *Client) Close() error {
	client.mu.Lock()
	defer client.mu.Unlock()
	if client.closing {
		return ErrShutDown
	}
	client.closing = true
	return client.cc.Close()
}

func (client *Client) IsAvailable() bool {
	client.mu.Lock()
	defer client.mu.Unlock()
	return !client.closing && !client.shutdown
}

func (client *Client) registerCall(call *Call) (uint64,error) {
	client.mu.Lock()
	defer client.mu.Unlock()
	if client.closing || client.shutdown {
		return 0, ErrShutDown
	}
	call.Seq = client.seq
	client.seq++
	client.pending[call.Seq] = call
	return call.Seq,nil
}

func (client *Client) removeCall(seq uint64) *Call {
	client.mu.Lock()
	defer client.mu.Unlock()
	call := client.pending[seq]
	delete(client.pending,seq)
	return call
}

func (client *Client) terminateCalls(err error)  {
	client.sending.Lock()
	defer client.sending.Unlock()
	client.mu.Lock()
	defer client.mu.Unlock()
	client.shutdown = true
	for _,call := range client.pending {
		call.Error = err
		call.done()
	}
}

/*
	处理接受函数
	每次处理一个call
	1.读取header
	2.判断seq是否在里面
 */
func (client *Client) receive()  {
	var err error
	for err == nil {
		var h codec.Header
		if err = client.cc.ReadHeader(&h); err != nil {
			break
		}
		call := client.removeCall(h.Seq)
		switch {
		case call==nil:
			err = client.cc.ReadBody(nil)
		case h.Error != "":
			call.Error = fmt.Errorf(h.Error)
			err = client.cc.ReadBody(nil)
			call.done()
		default:
			err = client.cc.ReadBody(call.Reply)
			if err != nil {
				call.Error = errors.New("reading body " + err.Error())
			}
			call.done()
		}
	}
	client.terminateCalls(err)
}

func NewClient(conn net.Conn,opt *Option) (*Client,error)  {
	f := codec.NewCodecFuncMap[opt.CodecType]
	if f == nil {
		err := fmt.Errorf("invalid codec type %s",opt.CodecType)
		log.Println("rpc client: codec error:",err)
		return nil,err
	}
	if err := json.NewEncoder(conn).Encode(opt); err != nil {
		log.Println("rpc client : options error: ",err)
		_ = conn.Close()
		return nil,err
	}
	return newClientCodec(f(conn),opt),nil
}

func newClientCodec(cc codec.Codec,opt *Option) *Client {
	client := &Client{
		seq: 1,
		cc: cc,
		opt: opt,
		pending: make(map[uint64]*Call),
	}
	go client.receive()
	return client
}

func parseOptions(opts ...*Option) (*Option,error) {
	if len(opts) == 0 || opts[0] == nil {
		return DefaultOption,nil
	}
	if len(opts) != 1 {
		return nil,errors.New("number of options is more than 1")
	}
	opt := opts[0]
	opt.MagicNumber = MagicNumber
	if opt.CodecType == "" {
		opt.CodecType = DefaultOption.CodecType
	}
	return opt,nil
}

// Dial 可变参数
func Dial(network,address string,opts ...*Option) (client *Client,err error) {
	return dialTimeout(NewClient,network,address,opts...)
}

func (client *Client) send(call *Call)  {
	client.sending.Lock()
	defer client.sending.Unlock()

	seq,err := client.registerCall(call)
	if err != nil {
		call.Error = err
		call.done()
		return
	}
	client.header.ServiceMethod = call.ServiceMethod
	client.header.Seq = seq
	client.header.Error = ""

	if err := client.cc.Write(&client.header,call.Args);err != nil {
		call := client.removeCall(seq)
		if call != nil {
			call.Error = err
			call.done()
		}
	}
}

func (client *Client) Go(serviceMethod	string,args,reply interface{},done chan *Call) *Call {
	if done == nil {
		done = make(chan *Call,1)
	} else if cap(done) == 0 {
		log.Panic("rpc client: done channel is unbuffered")
	}
	call := &Call{
		ServiceMethod: serviceMethod,
		Args: args,
		Reply: reply,
		Done: done,
	}
	client.send(call)
	return call
}

//func (client *Client) Call(serviceMethod string,args,reply interface{}) error {
//	call := <-client.Go(serviceMethod,args,reply,make(chan *Call,1)).Done
//	return call.Error
//}

func (client *Client) Call(ctx context.Context,serviceMethod string,args,reply interface{}) error {
	call := client.Go(serviceMethod,args,reply,make(chan *Call,1))
	select {
	case <- ctx.Done():
		client.removeCall(call.Seq)
		return errors.New("rpc client failed:"+ctx.Err().Error())
	case call := <-call.Done:
		return 	call.Error
	}
}

//连接超时--处理---给Dial加上一个超时外壳
type clientResult struct {
	client *Client
	err error
}
type newClientFunc func(conn net.Conn,opt *Option) (client *Client,err error)

func dialTimeout(f newClientFunc,network,address string,opts ...*Option) (client *Client,err error) {
	opt, err := parseOptions(opts...)

	if err != nil {
		return nil,err
	}
	conn,err := net.DialTimeout(network,address,opt.ConnectionTimeout)
	if err != nil {
		return nil,err
	}

	defer func() {
		if err != nil {
			_ = conn.Close()
		}
	}()

	ch := make(chan clientResult)
	go func() {
		client,err := f(conn,opt)
		ch <- clientResult{client: client,err:err}
	}()
	if opt.ConnectionTimeout == 0 {
		result := <- ch
		return result.client,result.err
	}
	select {
		case <- time.After(opt.ConnectionTimeout):
			return nil,fmt.Errorf("rpc client: connect timeout expect within %s",opt.ConnectionTimeout)
		case result := <- ch:
			return result.client,result.err
	}
}

func NewHTTPClient(conn net.Conn,opt *Option) (*Client,error) {
	_, _ = io.WriteString(conn,fmt.Sprintf("CONNECT %s HTTP/1.0\n\n",defaultRPCPath))

	resp,err := http.ReadResponse(bufio.NewReader(conn),&http.Request{Method: "CONNECT"})
	fmt.Println(resp.Status)
	if err == nil && resp.Status == connected {
		return NewClient(conn,opt)
	}
	if err == nil {
		err = errors.New("unexpected HTTP response: "+resp.Status)
	}
	return nil,err
}

func DialHTTP(network,address string,opts ...*Option) (*Client,error) {
	return dialTimeout(NewHTTPClient,network,address,opts...)
}

func XDial(rpcAddr string,opts ...*Option) (*Client,error) {
	parts := strings.Split(rpcAddr,"@")
	if len(parts) != 2 {
		return nil,fmt.Errorf("rpc client err: wrong format '%s' ,expect protocol@addr",rpcAddr)
	}
	protocol, addr := parts[0],parts[1]
	switch protocol {
	case "http":
		return DialHTTP("tcp",addr,opts...)
	default:
		return Dial(protocol,addr,opts...)
	}
}


